package com.zhss.dfs.backupnode.server;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

/**
 * 从namenode同步editlog组件
 */
public class EditLogFetcher extends Thread {

    private BackupNode backupNode;

    private NameNodeRpcClient namenode;

    private FSNameSystem nameSystem;

    private final Integer BACKUP_NODE_FETCH_SIZE = 10;

    public EditLogFetcher(BackupNode backupNode, FSNameSystem nameSystem, NameNodeRpcClient namenode) {
        this.backupNode = backupNode;
        this.namenode = namenode;
        this.nameSystem = nameSystem;

    }

    @Override
    public void run() {
        System.out.println("editlog抓取线程已启动");
        while (backupNode.isRunning()) {
            try {
                long syncedTxid = nameSystem.getSyncedTxid();
                JSONArray editlogs = this.namenode.fetchEditLog(syncedTxid);
                if (editlogs.size() == 0) {
                    System.out.println("没有拉取到任何一条数据，等待一秒后重新拉取");
                    Thread.sleep(1000);
                    continue;
                }
                if (editlogs.size() < BACKUP_NODE_FETCH_SIZE) {
                    Thread.sleep(1000);
                    System.out.println("拉取到的数据少于" + BACKUP_NODE_FETCH_SIZE + "条，等待1秒后继续拉取");
                }
                for (int i = 0; i < editlogs.size(); i++) {
                    JSONObject editlog = editlogs.getJSONObject(i);
                    System.out.println("拉取到一条editlog:" + editlog.toJSONString());
                    String op = editlog.getString("op");

                    if ("mkdir".equals(op)) {
                        String path = editlog.getString("path");
                        nameSystem.mkdir(editlog.getLong("txid"), path);

                    }
                }
                namenode.setIsNamenodeRunning(true);
            } catch (Exception e) {
                namenode.setIsNamenodeRunning(false);
            }
        }
    }
}
